dbt Cloud Orchestration で Fivetran から dbt Cloud のジョブを実行する #Fivetran

モダンデータスタック(MDS)

#Fivetran

#dbt

安原朋紀

2024.08.22

はじめに

Fivetran では dbt Cloud Orchestration として Fivetran 側から dbt Cloud のジョブをトリガー実行できる機能が提供されています。こちらの一通りの手順をまとめてみたく記事としました。

また、本機能については以下の記事でも紹介されていますので、こちらもあわせてご参照ください。

概要

公式ドキュメントとしては以下に記載があります。

dbt Cloud でもスケジュールなどでのジョブの実行が可能ですが、この機能では以下のいずれかをトリガーに Fivetran から dbt Cloud のジョブを実行できます。

カスタムスケジュール
- 指定の時間間隔で dbt Cloud のジョブをトリガーできる
統合スケジュール
- 選択したコネクタの同期をトリガーに dbt Cloud のジョブを実行

前提条件

dbt Cloud のプラン
- Team または Enterprise
- dbt Cloud Orchestration を使用するには、dbt Cloud への API アクセスが必要なため

https://www.getdbt.com/pricing

本機能の制約

1つのプロジェクトは、1つの宛先にしか設定できないため、同じプロジェクトを他の宛先に設定したい場合は、最初にそのプロジェクトを現在の宛先から削除する必要があります
統合スケジュールでは、ジョブをトリガーするすべてのコネクタは、同じ宛先に関連付けられている必要があります
コネクタの同期により、同期で新しいデータが到着したかどうかに関係なく、ジョブの実行自体はトリガーされます

事前準備

Fivetran 側

宛先には Snowflake を使用します。以下の手順で宛先として定義しておきました。

データソースには Google スプレッドシートを使用します。ここでは ChatGPT に作ってもらった以下のサンプルデータを使用します。

orders テーブル

order_id	customer_id	order_date	total_amount	product_id	quantity
1	101	2024-08-01	150	1	2
2	102	2024-08-03	100	2	1
3	103	2024-08-05	75	3	3
4	101	2024-08-07	300	1	4
5	104	2024-08-09	50	4	1
6	105	2024-08-10	500	2	5
7	106	2024-08-12	150	3	2
8	102	2024-08-13	230	4	3
9	103	2024-08-15	90	1	1
10	107	2024-08-18	250	2	2

customer テーブル

customer_id	customer_name	contact_email
101	Alice	alice@example.com
102	Bob	bob@example.com
103	Charlie	charlie@example.com
104	David	david@example.com
105	Eve	eve@example.com
106	Frank	frank@example.com
107	Grace	grace@example.com

それぞれをスプレッドシートの異なるシートに張り付け以下の手順でコネクタの設定を行います。

接続設定後、初期同期を行います。この時点で下図のようにテーブルが作成されます。

dbt Cloud 側

データロードが完了したら dbt Cloud 側で検証用にサンプルプロジェクトを作成し初期化を行います。models フォルダは以下の構成としました。

models
├── marts
│   └── total_spent_per_customer.sql
├── source.yml
└── staging
    ├── stg_google_sheets__customer.sql
    └── stg_google_sheets__order.sql

各ファイルは以下のように定義しています。

source.yml

version: 2

sources:
  - name: google_sheets
    database: fivetran_database
    schema: google_sheets  
    tables:
      - name: orders
      - name: customer

stg_google_sheets__order.sql

select
	*
from
    {{ source('google_sheets', 'orders') }}

stg_google_sheets__customer.sql

select
	*
from
    {{ source('google_sheets', 'customer') }}

total_spent_per_customer.sql

WITH customer_data AS (
    SELECT 
        customer_id, 
        customer_name
    FROM 
        {{ ref('stg_google_sheets__customer') }}
),
order_data AS (
    SELECT 
        customer_id, 
        total_amount
    FROM 
        {{ ref('stg_google_sheets__order') }}
),

final AS (

    SELECT 
        customer_data.customer_id, 
        customer_data.customer_name, 
        SUM(order_data.total_amount) AS total_spent
    FROM 
        customer_data
    JOIN 
        order_data ON customer_data.customer_id = order_data.customer_id
    GROUP BY 
        customer_data.customer_id, customer_data.customer_name
    ORDER BY 
        total_spent DESC

)

SELECT
	*
FROM 
	final

リネージグラフは下図のようになります。

あわせてdbt_project.yml でmatrs フォルダのモデルはテーブルとして作成されるように設定しておきました。

dbt_project.yml の一部

models:
  my_new_project:
    marts:
      +materialized: table

開発環境のスキーマでテスト後、環境とジョブを定義します。

環境の定義：prod スキーマに作成する設定

ジョブの設定

Fivetran 側でトリガー実行するため、ここでは dbt Cloud 側ではスケジュール等のトリガーの設定は行いませんでした。

セットアップ

事前準備ができたので、以降の手順で dbt Cloud Orchestration の設定を行います。ドキュメントとしては以下に記載があります。

dbt Cloud：サービストークンの取得

Fivetran から dbt Cloud のジョブをトリガーするために、トークンを取得します。dbt Cloud にログインし右上の歯車マークから「Account settings > Service tokens」をクリックします。

2024-08-22_12h06_20

トークンの一覧画面が開くので [+ Create service token] をクリックします。

任意のトークン名を入力し「Permission set」でトークンのパーミッションを設定します。ここでは dbt Cloud のプランに応じてそれぞれ以下の通り設定します。

Team プラン：Member
Enterprise プラン：Developer

あわせてトークンでアクセス可能なプロジェクトと環境を指定します。下図ではパーミッションとして「Developer」を選択しています。

設定を保存するとトークンが表示されるので、安全な場所に控えておきます。

Fivetran：dbt Cloud Orchestration の設定

Fivetran ダッシュボードの「Transformations」をクリックし、対象となる宛先を選択します。

はじめての場合、dbt Cloud のリージョン、トークンの登録画面が表示されるので、先の手順で取得した値を入力し [Authenticate] をクリックします。するとトークン権限でアクセス可能な dbt プロジェクトを選択できるので、ドロップダウンから選択します。

設定を保存するとプロジェクトへの接続テストが実行されます。

テストが終わると下図の表示になるので [Add Transforamtion] をクリックします。

どのタイプの変換を行うかを選択します。ここでは dbt Cloud 上のプロジェクトを使用した変換を行いたいので、dbt Cloud の [Add transormations] をクリックします。

すると下図の表示となり、対象のプロジェクト内でアクセス可能なジョブが表示されます。

先の手順で [Manage project] をクリックすると、接続解除などのプロジェクト設定を行えます。

トリガー実行したいジョブを選択すると、どのようなスケジュール実行とするか選択できます。ここでは、Fivetran の同期とあわせてジョブをトリガーしたいので、「Integrated」を選択します。

なお、「Custom」を選択すると指定の頻度でスケジュール実行が可能です。

次に、対象の宛先でジョブのトリガー条件となるコネクタを指定します。複数のコネクタを指定可能です。

コネクタを指定後 [Save and Run] をクリックして設定を完了です。同期のタイミングでトリガーされるため、設定完了後すぐにジョブがトリガーされることはありません。

コネクタの同期を実行

ここでは、手動でコネクタの同期を実行してみます。

同期開始後、すぐにジョブ側もを実行状態となっていました。

実行完了後は下図の表示となります。

Snowflake 側を確認するとジョブで指定した環境で設定したスキーマにオブジェクトが作成されています。

データも確認してみます。データマートを想定したテーブルは下図のように作成されています。（この時点で7レコード）

レコードを追加

さいごにデータソース側でレコードを追加してみます。

orders テーブル：青枠の３レコードを追加

customer テーブル：青枠の２レコードを追加

レコードを追加後、同期を実行します。実行後 Snowflake 側でそれぞれテーブルを確認します。

orders テーブル：赤枠のレコードが追加

customer テーブル：赤枠のレコードが追加

dbt Cloud との連携を実施しているので、同期と合わせてジョブも実行されます。

変換処理の結果（モデル）を見ると、ジョブが実行された結果こちらも更新されていることが確認できます。赤枠のレコードがデータソースの更新とあわせて追加・更新されています。

さいごに

Fivetran の dbt Cloud Orchestration で dbt Cloud のジョブをトリガーしてみました。Fivetran のデータロード機能と dbt Cloud の変換機能それぞれの強みを活用しつつ簡単に連携設定ができる機能なので組み合わせて使っていけると便利だなと感じました。こちらの内容が何かの参考になれば幸いです。